package com.wq.sofa.framework.bolt.server;

import com.alipay.remoting.Connection;
import com.alipay.remoting.ConnectionEventProcessor;
import com.alipay.remoting.ConnectionEventType;
import com.alipay.remoting.exception.RemotingException;
import com.alipay.remoting.rpc.RpcServer;
import com.alipay.remoting.rpc.protocol.SyncUserProcessor;
import com.wq.sofa.framework.bolt.ConnectEventProcessor;
import com.wq.sofa.framework.bolt.DisConnectEventProcessor;
import com.wq.sofa.framework.bolt.model.CommonRequest;
import io.netty.util.concurrent.DefaultThreadFactory;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Calendar;
import java.util.HashMap;
import java.util.Iterator;
import java.util.Map;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

/**
 * @Author: wangqiang20995
 * @Date:2018/9/19
 * @Description:
 * @Resource:
 */
public class SimpleSynServer {

    private RpcServer rpcServer;

    private int bindPort;

    private Map<Byte, SyncUserProcessor<CommonRequest>> processorMap;

    private ConnectEventProcessor connection;

    private DisConnectEventProcessor disConnection;

    private ExecutorService executorService;

    private Logger logger = LoggerFactory.getLogger(getClass());

    public SimpleSynServer(int port) {
        this.bindPort = port;
        this.connection = new ConnectEventProcessor();
        this.disConnection = new DisConnectEventProcessor();
        this.processorMap = new HashMap<>();
        this.rpcServer = new RpcServer(this.bindPort);
        this.executorService = new ThreadPoolExecutor(10, 100, 600, TimeUnit.SECONDS,
                new ArrayBlockingQueue<>(1000), new DefaultThreadFactory("RpcServer-Asyn-connection-pool"));
    }

    public void registerUserProcessor(Byte code, SyncUserProcessor<CommonRequest> userProcessor) {
        this.processorMap.put(code, userProcessor);
    }

    public void startServer() {
        if (processorMap == null || processorMap.size() == 0) {
            logger.error("not found or register processor for server,and please register them");
            return;
        }

        this.rpcServer.addConnectionEventProcessor(ConnectionEventType.CONNECT, connection);
        this.rpcServer.addConnectionEventProcessor(ConnectionEventType.CLOSE, disConnection);
        Iterator<Map.Entry<Byte, SyncUserProcessor<CommonRequest>>> iterator = this.processorMap.entrySet().iterator();
        while (iterator.hasNext()) {
            SyncUserProcessor<CommonRequest> processor = iterator.next().getValue();
            this.rpcServer.registerUserProcessor(processor);
        }

        this.rpcServer.start();
        logger.info("SimpleSynServer started...");
    }

    public void stopServer() {
        logger.info("SimpleSynServer stop...");
        this.rpcServer.stop();
    }

    public void transportInConnection(Connection connection, int times) {
        initiativeSendByServer(times, connection);
    }

    private void initiativeSendByServer(int times, Connection toClient) {
        if (toClient != null) {
            try {
                for (int i = 0; i < times; i++) {
                    CommonRequest commonRequest = new CommonRequest();
                    commonRequest.setSystem("Hello client,and this is sent by server initiative[" + (i + 1) + "]");
                    this.rpcServer.invokeSync(toClient, commonRequest, 10000);
                }
            } catch (Exception e) {
                logger.error("send message to client initiative failed and cause is :" + e.getMessage(), e);
            }
        }
    }


    public ExecutorService getExecutorService() {
        return executorService;
    }
}
